package com.corn.flink.lesson3;

import org.apache.flink.api.common.serialization.Encoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.TimeUnit;

/**
 * sinkOperator
 *
 * @author JimWu
 * @date 2023/2/28 16:41
 **/
public class FlinkFileSinkOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Person> ds = env.fromElements(
                new Person("alice", 15, "cq"),
                new Person("jack", 18, "cq"),
                new Person("rose", 20, "sh"),
                new Person("mick", 42, "gz"),
                new Person("white", 12, "cq"),
                new Person("maria", 32, "sz"),
                new Person("candy", 56, "bj"),
                new Person("jerry", 28, "wh")
        );
        // 1. 标准输出
//        ds.print();
        // 2. 输出到文件系统
        StreamingFileSink.RowFormatBuilder fileSink = StreamingFileSink.<Person>forRowFormat(
                new Path("./output"),
                new Encoder<Person>() {
                    @Override
                    public void encode(Person element, OutputStream stream) throws IOException {
                        stream.write(element.toString().getBytes());
                    }
                }
        ).withRollingPolicy(DefaultRollingPolicy.builder()
                // 文件滚动最大值 单位字节
                .withMaxPartSize(1024*1024*1024)
                // 闲置最大时间
                .withInactivityInterval(TimeUnit.MINUTES.toMillis(15))
                // 滚动间隔时间
                .withRolloverInterval(TimeUnit.MINUTES.toMillis(20))
                .build());
        ds.addSink(fileSink.build());
        env.execute();
    }
}
